package com.ibm.cps.storm;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;

import backtype.storm.task.OutputCollector;

import com.ibm.cps.spark.streaming.TestTuple;
import com.ibm.util.TopologyStreamIds;

public class DispatchBoltCollector extends OutputCollector {

	HashSet<String> destination = new HashSet<String>();
	ArrayList<TestTuple> commitTuples = new ArrayList<TestTuple>();

	// private static final Logger logger = LoggerFactory
	// .getLogger(DispatchBoltCollector.class);

	public DispatchBoltCollector() {
		super(null);
	}

	@Override
	public List<Integer> emit(String streamId, List<Object> tuple) {
		if (streamId.equals(TopologyStreamIds.DATAMESSAGE_STREAM)) {
			destination.add((String) (tuple.get(0)));
		}
		commitTuples.add(new TestTuple(streamId, tuple));
		return null;
	}

	@Override
	public List<Integer> emit(List<Object> tuple) {
		return null;
	}

	public HashSet<String> getDestination() {
		return destination;
	}

	public ArrayList<TestTuple> getCommitTuples() {
		return commitTuples;
	}

	public void clearCommitTuples() {
		commitTuples.clear();
	}
}
